package com.zhangdi.mgsync.core.thread;

import com.mongodb.MongoCommandException;
import com.mongodb.client.ChangeStreamIterable;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoCursor;
import com.mongodb.client.model.changestream.ChangeStreamDocument;
import com.mongodb.client.model.changestream.FullDocument;
import org.bson.BsonDocument;
import org.bson.Document;
import org.bson.conversions.Bson;

import java.util.List;

/**
 * Created by zhangdi on 2020/07/16 20:36:13.
 */
public class WatchThread implements Runnable {
    private MongoCollection<Document> collection;
    private BsonDocument resumeToken;
    private List<Bson> pipeline;

    public WatchThread(MongoCollection<Document> collection, BsonDocument resumeToken, List<Bson> pipeline) {
        this.collection = collection;
        this.resumeToken = resumeToken;
        this.pipeline = pipeline;
    }

    @Override
    public void run() {
        MongoCursor<ChangeStreamDocument<Document>> cursor = null;
        ChangeStreamIterable<Document> iterable = collection.watch().fullDocument(FullDocument.UPDATE_LOOKUP).resumeAfter(resumeToken);
        try {
            cursor = iterable.iterator();
        } catch (MongoCommandException e) {
            cursor = collection.watch(pipeline).fullDocument(FullDocument.UPDATE_LOOKUP).iterator();
        }

        while (cursor.hasNext()) {
            ChangeStreamDocument<Document> document = cursor.next();
            BsonDocument re = document.getResumeToken();
            System.out.println(re + "   " + document.getFullDocument());
        }
    }
}
